package com.feishi.core.util.queue.feishi;



import com.feishi.core.util.queue.Customer;
import com.feishi.core.util.queue.Producer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by caixq on 4/26 026.
 */
public class FeishiProducer<T,R> implements Producer<Msg<T,R>> {
    Map<T,List<FeishiExecutor<T,R>>> executorMap=new HashMap<T, List<FeishiExecutor<T,R>>>();

    private ExecutorService pool;
    private Customer<Msg<T,R>> customer;
    public void push(Msg<T,R> msg) {
        T key = msg.getGroup();
        List<FeishiExecutor<T,R>> executors=executorMap.get(key);
        if(executors==null) {
            executors = new ArrayList<>();
            executorMap.put(key,executors);
        }
        FeishiExecutor<T, R> executor;
        if(executors.size()==0){
            executor = new FeishiExecutor<>(customer);
            executor.add(msg);
            executors.add(executor);
            execute(executors);
        }else{
            synchronized (executors){
                executor=executors.get(executors.size()-1);
                if(executor.execStatus==2){
                    executor= new FeishiExecutor<T, R>(customer);
                    executors.add(executor);
                }
                executor.add(msg);

            }
        }
    }

    private void execute(List<FeishiExecutor<T, R>> executors) {
        if(executors==null||executors.size()==0){
            return;
        }
        pool.execute(()->{
            FeishiExecutor<T, R> executor = executors.get(0);
            executor.run();
            synchronized (executors){
                executors.remove(executor);
            }
            execute(executors);
        });
    }

    public void execute(){

    }

    public FeishiProducer(Customer<Msg<T,R>> customer) {
        this.customer = customer;
        this.pool= Executors.newCachedThreadPool();
    }

    public FeishiProducer( Customer<Msg<T, R>> customer,ExecutorService pool) {
        this.pool = pool;
        this.customer = customer;
    }

    public void close() {
        pool.shutdown();

    }

}
